#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcomm/mq_log.hpp"
#include "../mqcomm/mq_helper.hpp"
#include "../mqcomm/mq_msg.pb.h"
#include "../mqcomm/mq_proto.pb.h"
#include "../mqcomm/mq_threadpool.hpp"

#include "mq_consumer.hpp"

#include <mutex>
#include <condition_variable>
#include <unordered_map>

namespace ns_kkymq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(muduo::net::TcpConnectionPtr conn, const ProtobufCodecPtr &codec)
            : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec)
        {
        }

        ~Channel()
        {
            basicCancel();
        }

        bool openChannel()
        {
            openChannelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);

            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void closeChannel()
        {
            closeChannelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            waitResponse(rid);
        }

        std::string cid()
        {
            return _cid;
        }

        bool declareExchange(const std::string &name, ExchangeType type, bool durable,
                             bool auto_delete, google::protobuf::Map<std::string, std::string> &args)
        {
            // 1.构造一个声明虚拟机的请求对象
            std::string rid = UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // std::cout<<"cid : "<<_cid<<std::endl;
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 2.向服务器发送请求
            _codec->send(_conn, req);
            // 3.等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            // 4.返回
            return resp->ok();
        }
        void deleteExchange(const std::string &name)
        {
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            _codec->send(_conn, req);
            // 等待服务器的响应
            waitResponse(rid);
            return;
        }

        bool declareQueue(const std::string &name, bool durable, bool exclusive, bool auto_delete,
                          google::protobuf::Map<std::string, std::string> &args)
        {
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(name);
            req.set_durable(durable);
            req.set_excludsive(exclusive);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            _codec->send(_conn, req);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            // 返回
            return resp->ok();
        }

        void deleteQueue(const std::string &name)
        {
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(name);
            _codec->send(_conn, req);
            // 等待服务器响应
            waitResponse(rid);
        }

        bool queueBind(const std::string &e_name, const std::string &q_name, const std::string key)
        {
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(e_name);
            req.set_queue_name(q_name);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        bool queueUnBind(const std::string &e_name, const std::string &q_name)
        {
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(e_name);
            req.set_queue_name(q_name);
            _codec->send(_conn, req);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void basicPublish(const std::string &e_name, const BasicProperties *bp, const std::string &body)
        {
            std::string rid = UUIDHelper::uuid();
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(e_name);
            req.set_body(body);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }

        void basicAck(const std::string &msgid)
        {
            if (_consumer.get() == nullptr)
            {
                lg(Debug,"消息确认时，找不到消费者的信息!");
                return;
            }
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_q_name);
            req.set_message_id(msgid);
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }

        void basicCancel()
        {
            if (_consumer.get() == nullptr)
            {
                // lg(Debug,"取消订阅时,找不到消费者信息");
                return;
            }
            std::string rid = UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->_tag);
            req.set_queue_name(_consumer->_q_name);
            _codec->send(_conn, req);
            waitResponse(rid);
            _consumer.reset();
            return;
        }

        bool basicConsume(const std::string &consumer_tag, const std::string &q_name,
                          bool auto_ack, const ConsumerCallback &cb)
        {
            if (_consumer.get() != nullptr)
            {
                lg(Debug, "当前信道已订阅其他队列消息!");
                return false;
            }
            std::string rid = UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(q_name);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);
            // lg(Debug,"_codec发送请求成功");
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok() == false)
            {
                lg(Debug, "添加订阅失败!");
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag, q_name, auto_ack, cb);
            // lg(Debug, "成功添加订阅，消费者标签: %s", consumer_tag.c_str());
            return true;
        }

        // 连接收到基础响应，向hash_map中添加响应
        void putBasicResponse(const basicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _basic_resp.insert({resp->rid(), resp});
            _cv.notify_all();
        }
        // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        void consume(const basicConsumeResponsePtr &resp)
        {
            // lg(Debug, "resp_tag:%s,resp_cid:%s,resp_body:%s",resp->consumer_tag().c_str(),resp->cid().c_str(),resp->body().c_str());
            if (_consumer.get() == nullptr)
            {
                lg(Debug, "消息处理时，未找到订阅者信息!");
                return;
            }
            if (_consumer->_tag != resp->consumer_tag())
            {
                lg(Debug, "收到的推送消息中的消费者标识,与当前信道消费者标识不一致!now:%s,receive:%s",_consumer->_tag.c_str(),resp->consumer_tag().c_str());
                return;
            }
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }

    private:
        basicCommonResponsePtr waitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mtx);

            // 伪代码是 while(condition()) _cv.wait(lock)
            // 也就是被唤醒后需要先去判断条件是否满足，然后再决定是否继续等待
            // lg(Debug, "wait开始");           注意 没收到响应可能是服务器没有发送response 
            _cv.wait(lock, [this, &rid]()
                     { return _basic_resp.find(rid) != _basic_resp.end(); });
            // lg(Debug, "wait结束");

            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }

    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer;
        std::mutex _mtx;
        std::condition_variable _cv;
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {
        }

        Channel::ptr create(muduo::net::TcpConnectionPtr &conn, ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto channel = std::make_shared<Channel>(conn, codec);
            _channels.insert({channel->cid(), channel});
            return channel;
        }

        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _channels.erase(cid);
        }

        Channel::ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mtx;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}